/**
 * 
 */

package com.wutianyi.taskqueue.queue.persistent.impl;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.log4j.Logger;

import com.wutianyi.taskqueue.constants.TaskStatus;
import com.wutianyi.taskqueue.dao.TaskDao;
import com.wutianyi.taskqueue.dataobject.TaskQueueDO;
import com.wutianyi.taskqueue.queue.persistent.DBPersistent;

/**
 * @author wutianyi
 * 
 */
public class DBPersistentImpl implements DBPersistent<TaskQueueDO> {

	private static Logger							logger					=
																					Logger
																							.getLogger(DBPersistentImpl.class);
	private TaskDao<TaskQueueDO>					taskDao;

	private final static int						MAX_WORK_THREAD_SIZE	= 16;
	/**
	 * 批量持久化数量
	 */
	private final static int						MAX_BATCH_INSERT_SIZE	= 500;

	private int										batchInsertSize;

	/**
	 * 持久化的线程的数目
	 */
	private int										workThreadSize;

	/**
	 * 工作的线程
	 */
	private Thread[]								workThreads;

	private boolean									isRunning;

	private BlockingQueue<TaskQueueDO>				queue;
	/**
	 * 
	 */
	private Listener								listener;

	/**
	 * 对待插入的任务建立索引
	 */
	private ConcurrentHashMap<String, TaskQueueDO>	indexs;

	// FIXME!删除
	public AtomicInteger							insertCount				= new AtomicInteger(0);

	public DBPersistentImpl(int wordThreadSize, int batchInsertSize) {

		this.workThreadSize = wordThreadSize;
		this.batchInsertSize = batchInsertSize;
		this.indexs = new ConcurrentHashMap<String, TaskQueueDO>(10000);
		workThreads = new Thread[this.workThreadSize];
		queue = new LinkedBlockingQueue<TaskQueueDO>();
		listener = new Listener();
	}

	public DBPersistentImpl() {

		this(MAX_WORK_THREAD_SIZE, MAX_BATCH_INSERT_SIZE);
	}

	/**
	 * 开始执行任务
	 */
	public void start() {

		isRunning = true;
		for(int i = 0; i < workThreadSize; i++) {
			workThreads[i] = new Thread(listener);
			workThreads[i].start();
		}
	}

	/**
	 * 关闭任务
	 */
	public void shutDown() {

		isRunning = false;
		for(Thread thread : workThreads) {
			if(null != thread && thread.isAlive()) {
				thread.interrupt();
			}
		}

		for(Thread thread : workThreads) {
			try {
				thread.join();
			}catch(InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @seecom.duowan.zone.taskqueue.queue.persistent.Persistent#addTask(com.duowan.zone.taskqueue.
	 * dataobject.TaskQueueDO)
	 */
	@Override
	public boolean addTask(TaskQueueDO task) {

		indexs.put(task.getTaskId(), task);
		return queue.offer(task);
	}

	public boolean updateTaskStatusFromIndex(String taskId, int status) {

		TaskQueueDO task = indexs.get(taskId);

		if(null != task) {
			task.setStatus(status);
		}

		if(indexs.contains(taskId)) {
			return true;
		}

		return false;
	}

	public boolean removeTaskFromIndex(TaskQueueDO task) {

		TaskQueueDO t = indexs.remove(task.getTaskId());
		if(null == t) {
			return false;
		}
		return true;
	}

	private void batchRemoveTaskFromIndex(List<TaskQueueDO> tasks) {

		if(null == tasks || 0 == tasks.size()) {
			return;
		}
		Iterator<TaskQueueDO> itr = tasks.iterator();
		while(itr.hasNext()) {
			TaskQueueDO task = itr.next();
			if(!removeTaskFromIndex(task)) {
				itr.remove();
			}
		}
	}

	public void setTaskDao(TaskDao<TaskQueueDO> taskDao) {

		this.taskDao = taskDao;
	}

	public Thread[] getWorkThreads() {

		return workThreads;
	}

	/**
	 * 获取当前队列中待持久化的任务数
	 * 
	 * @return
	 */
	public int getQueueSize() {

		return queue.size();
	}

	/**
	 * 获取任务索引中的数量，size跟queueSize不一定保持一致
	 * 
	 * @return
	 */
	public int getIndexSize() {

		return indexs.size();
	}

	private class Listener implements Runnable {

		/*
		 * (non-Javadoc)
		 * 如果带持久化的队列中的task超过batchInsertSize，那么就批量插入
		 * 
		 * @see java.lang.Runnable#run()
		 */
		@Override
		public void run() {

			while(isRunning) {
				try {
					if(queue.size() >= batchInsertSize) {
						List<TaskQueueDO> tasks = new ArrayList<TaskQueueDO>();
						queue.drainTo(tasks, batchInsertSize);
						batchRemoveTaskFromIndex(tasks);
						taskDao.batchAddTasks(tasks, TaskStatus.NEW.getValue());
						insertCount.addAndGet(tasks.size());
					}else {

						TaskQueueDO task = queue.take();
						if(removeTaskFromIndex(task)) {

							taskDao.addTask(task);
							insertCount.addAndGet(1);
						}
					}
				}catch(InterruptedException e) {
					// FIXME!
				}
			}
		}

	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.duowan.zone.taskqueue.queue.persistent.Persistent#batchAddTasks(java.util.List)
	 */
	@Override
	public void batchAddTasks(List<TaskQueueDO> tasks) {

		for(TaskQueueDO task : tasks) {
			addTask(task);
		}
	}
}
